package com.captjack.bigdata.hadoop.hbase.nature;

import com.captjack.common.util.thread.ThreadPoolFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;

/**
 * Create by Capt Jack
 * 获取HBase连接，操作HBase连接，此处只创建连接，并不管理连接
 *
 * @author Administrator
 */
public class HBaseConnection {

    /**
     * 日志
     */
    private static final Logger logger = LoggerFactory.getLogger(HBaseConnection.class);

    /**
     * HBase连接
     * hBase的单个连接可以被Table和Admin实例共享，连接创建是一个重量级的操作。
     * 连接实现是线程安全的，客户端可以创建一次，连接并分享给不同的线程。
     * Table和Admin是轻量级的而且不是线程安全的。通常情况下，
     * 每个客户端应用程序的单个连接实例化，并且每个线程将获取它自己的表实例。
     * 缓存或线程池Table和Admin不推荐使用。
     * 一个应用（进程）对应着一个connection，
     * 每个应用里的线程通过调用connection的getTable方法从connection维护的线程池里获得table实例，
     * 按官方的说法，这种方式获得的table是线程安全的。
     * 每次table读写之后应该把table close掉，整个进程结束的时候才把connection close掉。
     */
    private static Connection connection = null;

    /**
     * 线程池的大小
     */
    private static final int THREAD_POOL_SIZE = 100;

    /**
     * hadoop配置
     */
    private static final Configuration CONFIGURATION = HBaseConfiguration.create();

    /**
     * 创建一个定长的线程池
     */
    private static final ExecutorService EXECUTOR_SERVICE = ThreadPoolFactory.newFixedThreadPool(THREAD_POOL_SIZE, "HbaseConnection");

    static {
        CONFIGURATION.set("hbase.zookeeper.quorum", "host192,host194,host195");
        CONFIGURATION.set("hbase.client.scanner.caching", "500");
        // 创建连接
        try {
            connection = ConnectionFactory.createConnection(CONFIGURATION, EXECUTOR_SERVICE);
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("get HBase connection failed ! cause by : " + e);
        }
        // 如果创建连接失败，那么直接结束进程，退出程序
        if (connection == null) {
            logger.error("get HBase connection failed!");
            System.exit(0);
        }
    }

    /**
     * 获取连接
     *
     * @return HBase连接
     */
    public static Connection getConnection() {
        return connection;
    }

    /**
     * 关闭HBase连接
     */
    public static void closeConnection(Connection connection) {
        try {
            if (connection != null) {
                connection.close();
            }
        } catch (Exception e) {
            logger.error("close HBase connection failed ! cause by : " + e);
        }
    }

}
